package com.alibaba.nacos.naming.core.v2.event.publisher;

import com.alibaba.nacos.common.notify.Event;
import com.alibaba.nacos.common.notify.ShardedEventPublisher;
import com.alibaba.nacos.common.notify.listener.Subscriber;
import com.alibaba.nacos.common.utils.ConcurrentHashSet;
import com.google.common.base.Preconditions;
import lombok.extern.slf4j.Slf4j;

import java.util.Collections;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
public class NamingEventPublisher
        extends Thread
        implements ShardedEventPublisher {

    private volatile boolean initialized = false;
    private volatile boolean shutdown = false;
    private BlockingQueue<Event> queue;
    private int queueMaxSize = -1;
    private String publisherName;
    private final Map<Class<? extends Event>, Set<Subscriber<? extends Event>>> subscribersMap = new ConcurrentHashMap<>();

    @Override
    public void init(Class<? extends Event> type,
                     int bufferSize) {
        this.queueMaxSize = bufferSize;
        this.queue = new ArrayBlockingQueue<>(bufferSize);
        this.publisherName = type.getSimpleName();
        super.setName("naming.publisher-" + publisherName);
        super.setDaemon(true);
        super.start();
        this.initialized = true;
    }

    @Override
    public boolean publish(Event event) {
        Preconditions.checkState(initialized, "Publisher does not start");
        boolean offered = this.queue.offer(event);
        if (!offered) {
            log.warn("publisher:{} 队列过载 无法添加时间 event：{}", publisherName, event);
            handleEvent(event);
        }
        return true;
    }

    @Override
    public void run() {
        try {
            handleEvents();
        } catch (Exception e) {
            log.error("publisher:{} occur exception", publisherName, e);
        }
    }

    private void handleEvents() {
        while (!shutdown) {
            try {
                Event event = queue.take();
                handleEvent(event);
            } catch (Exception e) {
                log.error("publisher:{} occur exception", publisherName, e);
                Thread.currentThread().interrupt();
            }
        }
    }

    private void handleEvent(Event event) {
        Class<? extends Event> eventType = event.getClass();
        Set<Subscriber<? extends Event>> subscribers = subscribersMap.get(eventType);
        if (subscribers == null) {
            log.warn("publisher:{} subscribers not found", publisherName);
            return;
        }

        for (Subscriber subscriber : subscribers) {
            notifySubscriber(subscriber, event);
        }
    }

    private void notifySubscriber(Subscriber subscriber,
                                  Event event) {
        subscriber.executor().execute(() -> subscriber.onEvent(event));
    }

    @Override
    public void addSubscriber(Subscriber subscriber) {
        addSubscriber(subscriber, subscriber.subscribeType());
    }

    @Override
    public void addSubscriber(Subscriber subscriber,
                              Class<? extends Event> subscribeType) {
        subscribersMap.computeIfAbsent(subscribeType, k -> new ConcurrentHashSet<>()).add(subscriber);
    }
}
